package com.twoandone.tms.realtime.app.dwd;

import com.twoandone.tms.realtime.utils.KafkaUtil;
import com.twoandone.tms.realtime.utils.MySqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class DwdWaybillOrderTrace {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment STE = StreamTableEnvironment.create(env);

        STE.executeSql(KafkaUtil.getTopicDbDDL("Dwd_Waybill_Order_Trace"));
        STE.getConfig().setIdleStateRetention(Duration.ofSeconds(15 * 60 + 5));

        Table orderTrace = STE.sqlQuery("select\n" +
                "         `data`['id'] id,\n" +
                "         `data`['order_id'] order_id,\n" +
                "         `data`['trace_desc'] trace_desc,\n" +
                "         `data`['create_time'] create_time,\n" +
                "         `data`['cur_task_id'] cur_task_id,\n" +
                "         `data`['task_type'] task_type,\n" +
                "         `data`['charge_emp_id'] charge_emp_id,\n" +
                "         `data`['is_rollback'] is_rollback,\n" +
                "         `data`['update_time'] update_time,\n" +
                "         `data`['is_deleted'] is_deleted,\n" +
                "          proc_time,\n" +
                "          ts\n" +
                "          from topic_db\n" +
                "          where `table` = 'order_trace_log'\n" +
                "          and `type` = 'insert'");
        STE.createTemporaryView("orderTrace", orderTrace);
        // STE.executeSql("select * from orderTrace").print();

        Table orderInfo = STE.sqlQuery("select \n" +
                "data['id'] id,\n" +
                "data['order_no'] order_no,\n" +
                "data['status'] status,\n" +
                "data['collect_type'] collect_type,\n" +
                "data['user_id'] user_id,\n" +
                "data['receiver_complex_id'] receiver_complex_id,\n" +
                "data['receiver_province_id'] receiver_province_id,\n" +
                "data['receiver_city_id'] receiver_city_id,\n" +
                "data['receiver_district_id'] receiver_district_id,\n" +
                "data['receiver_address'] receiver_address,\n" +
                "data['receiver_name'] receiver_name,\n" +
                "data['receiver_phone'] receiver_phone,\n" +
                "data['receive_location'] receive_location,\n" +
                "data['sender_complex_id'] sender_complex_id,\n" +
                "data['sender_province_id'] sender_province_id,\n" +
                "data['sender_city_id'] sender_city_id,\n" +
                "data['sender_district_id'] sender_district_id,\n" +
                "data['sender_address'] sender_address,\n" +
                "data['sender_name'] sender_name,\n" +
                "data['sender_phone'] sender_phone,\n" +
                "data['send_location'] send_location,\n" +
                "data['payment_type'] payment_type,\n" +
                "data['cargo_num'] cargo_num,\n" +
                "data['amount'] amount,\n" +
                "data['estimate_arrive_time'] estimate_arrive_time,\n" +
                "data['distance'] distance,\n" +
                "ts,\n" +
                "proc_time\n" +
                "from `topic_db` \n" +
                "where `table` = 'order_info'\n" +
                "and `type`= 'insert'");
        STE.createTemporaryView("orderInfo", orderInfo);
        // STE.executeSql("select * from orderInfo").print();

        STE.executeSql(MySqlUtil.getBaseDicLookUpDDL());
        // STE.executeSql("select * from base_dic").print();

        Table lookUpJoin = STE.sqlQuery("select  " +
                "            ot.id,\n" +
                "            ot.order_id,\n" +
                "            ot.trace_desc,\n" +
                "            ot.create_time,\n" +
                "            ot.cur_task_id,\n" +
                "            ot.task_type,\n" +
                "            ot.charge_emp_id,\n" +
                "            ot.is_rollback,\n" +
                "            ot.update_time,\n" +
                "            ot.is_deleted,\n" +
                "            oi.order_no,\n" +
                "            oi.status,\n" +
                "            oi.collect_type,\n" +
                "            oi.user_id,\n" +
                "            oi.receiver_complex_id,\n" +
                "            oi.receiver_province_id,\n" +
                "            oi.receiver_city_id,\n" +
                "            oi.receiver_district_id,\n" +
                "            oi.receiver_address,\n" +
                "            oi.receiver_name,\n" +
                "            oi.receiver_phone,\n" +
                "            oi.receive_location,\n" +
                "            oi.sender_complex_id,\n" +
                "            oi.sender_province_id,\n" +
                "            oi.sender_city_id,\n" +
                "            oi.sender_district_id,\n" +
                "            oi.sender_address,\n" +
                "            oi.sender_name,\n" +
                "            oi.sender_phone,\n" +
                "            oi.send_location,\n" +
                "            oi.payment_type,\n" +
                "            oi.cargo_num,\n" +
                "            oi.amount,\n" +
                "            oi.estimate_arrive_time,\n" +
                "            oi.distance,\n" +
                "            type_dic.name task_type_name,\n" +
                "            reason_dic.name order_status_name\n" +
                "            from " +
                "" +
                "orderTrace ot\n" +
                "            join \n" +

                "            orderInfo oi\n" +
                "            on ot.order_id = oi.id\n" +
                "            join \n" +

                "            base_dic for system_time as of ot.proc_time as type_dic\n" +
                "            on CAST( ot.task_type AS bigint) = type_dic.id\n" +
                "            join\n" +
                "            base_dic for system_time as of oi.proc_time as reason_dic\n" +
                "            on CAST( oi.status AS bigint) = reason_dic.id");
        STE.createTemporaryView("lookUpJoin", lookUpJoin);

        // STE.executeSql("select * from lookUpJoin").print();

        //
        //
        STE.executeSql("CREATE TABLE dwd_waybill_order_trace (\n" +
                "                    id string,\n" +
                "                    order_id string,\n" +
                "                    trace_desc string,\n" +
                "                    create_time string,\n" +
                "                    cur_task_id string,\n" +
                "                    task_type string,\n" +
                "                    charge_emp_id string,\n" +
                "                    is_rollback string,\n" +
                "                    update_time string,\n" +
                "                    is_deleted string,\n" +
                "                    order_no string,\n" +
                "                    status string,\n" +
                "                    collect_type string,\n" +
                "                    user_id string,\n" +
                "                    receiver_complex_id string,\n" +
                "                    receiver_province_id string,\n" +
                "                    receiver_city_id string,\n" +
                "                    receiver_district_id string,\n" +
                "                    receiver_address string,\n" +
                "                    receiver_name string,\n" +
                "                    receiver_phone string,\n" +
                "                    receive_location string,\n" +
                "                    sender_complex_id string,\n" +
                "                    sender_province_id string,\n" +
                "                    sender_city_id string,\n" +
                "                    sender_district_id string,\n" +
                "                    sender_address string,\n" +
                "                    sender_name string,\n" +
                "                    sender_phone string,\n" +
                "                    send_location string,\n" +
                "                    payment_type string,\n" +
                "                    cargo_num string,\n" +
                "                    amount string,\n" +
                "                    estimate_arrive_time string,\n" +
                "                    distance string,\n" +
                "                    task_type_name string,\n" +
                "                    order_status_name string,\n" +
                "                    PRIMARY KEY (id) NOT ENFORCED )"
                + KafkaUtil.getUpsertKafkaDDL("dwd_waybill_order_trace"));

        //写入kafka
        STE.executeSql("insert into dwd_waybill_order_trace select * from lookUpJoin");


    }
}
